package dacp.etl.kafka.hdfs.connect.writer;

import java.io.BufferedWriter;
import java.io.IOException;
import java.io.OutputStreamWriter;

import org.apache.commons.io.IOUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.Path; 
import org.apache.kafka.connect.sink.SinkRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import dacp.etl.kafka.hdfs.connect.utils.CommonSinkConnectorConfig;
import dacp.etl.kafka.hdfs.process.ProcessChain;
import io.confluent.connect.hdfs.RecordWriter;

public class RecordWriterProvider {
	private static final Logger log = LoggerFactory.getLogger(RecordWriterProvider.class);
	private final static String EXTENSION = ".AVL";
	private static final String lineSeparator = "\n"; 
	private SinkRecord record;
	private String fileName;

	public String getExtension() {
		return EXTENSION;
	}

	public SinkRecord getRecord() {
		return record;
	}
	
	public String getFileName(){
		return fileName;
	}

	public RecordWriter<SinkRecord> getRecordWriter(Configuration conf, String fileName, SinkRecord record,
			CommonSinkConnectorConfig config) throws IOException {

		Path path = new Path(fileName);
		log.info("getRecordWriter: " + path);
		final FSDataOutputStream out = path.getFileSystem(conf).create(path, false);
		
		final BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(out));
		this.record = record;
		this.fileName = fileName;
		
		final ProcessChain pc = ProcessChain.build(config);

		if(pc == null){
			return new RecordWriter<SinkRecord>() {
				@Override
				public void write(SinkRecord record) throws IOException {
					writer.write(record.value().toString() + lineSeparator);
				}
				@Override
				public void close() throws IOException {
					 IOUtils.closeQuietly(writer);
				}
			};
		}
		
		return new RecordWriter<SinkRecord>() {
			@Override
			public void write(SinkRecord record) throws IOException {
				writer.write(pc.process(record.value().toString()) + lineSeparator);
			}
			@Override
			public void close() throws IOException {
				IOUtils.closeQuietly(writer);
			}
		};

	}

}
